package com.yx.es.sync.sevice;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.yx.es.sync.dto.CdcDTO;
import com.yx.es.sync.dto.FullDataSyncDTO;
import com.yx.es.sync.mapper.DynamicTableMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author lsx
 * @date 2025/3/3 15:05
 */
@Slf4j
@Service
public class FullDataSyncService {

    private final ConcurrentHashMap<String, Boolean> runningTasks = new ConcurrentHashMap<>();

    @Resource
    private DynamicTableMapper dynamicTableMapper;

    @Async("taskExecutor")
    public void syncData(String nameServerAddr, FullDataSyncDTO taskDTO) {
        // 防止重复执行
        if (runningTasks.putIfAbsent(taskDTO.getTable(), true) != null) {
            System.out.println("任务已存在，跳过执行：" + taskDTO.getTable());
            return;
        }

        try {
            DefaultMQProducer producer = new DefaultMQProducer("Producer_" + taskDTO.getTable());
            // 配置 NameServer 地址
            producer.setNamesrvAddr(nameServerAddr);
            // 设置消息发送超时时间为 5 秒
            producer.setSendMsgTimeout(10000);
            // 设置发送失败时的重试次数
            producer.setRetryTimesWhenSendFailed(10);
            // 启动生产者
            producer.start();


            process(producer, taskDTO);

        } catch (Exception e) {
            log.error("[FullDataSyncService.syncData]发生错误：{}", e);
        } finally {
            // 任务执行完后移除
            runningTasks.remove(taskDTO.getTable());
        }
    }

    private Long result_succ_count = 0L;
    private Long result_error_count = 0L;

    private void sendMsg(DefaultMQProducer producer, String topic, String table, CdcDTO cdcDTO){
        String msg = JSONUtil.toJsonStr(cdcDTO);

        // 创建消息对象
        Message message = new Message(
                topic,
                "TAG_" + table,
                msg.getBytes());
        try {
            SendResult sendResult = producer.send(message);

            if (!Objects.equals(sendResult.getSendStatus(), SendStatus.SEND_OK)){
                throw new RuntimeException(StrUtil.format("消息发送失败：{}", msg));
            }else {
                log.info("消息发送成功: {} = {}", sendResult.getMsgId(), msg);
                result_succ_count++;
            }
        } catch (Exception e) {
            result_error_count++;
            log.error("[FullDataSyncService.sendMsg]消息发送失败：{}", msg);
            throw new RuntimeException(e);
        }finally {
            log.info("成功失败：{} = {}", result_succ_count, result_error_count);
        }

    }


    private void process(DefaultMQProducer producer, FullDataSyncDTO taskDTO){
        Long maxId = Long.parseLong(taskDTO.getMaxId());
        Long beginId = Long.parseLong(taskDTO.getMinId());
        while (beginId < maxId){
            List<Map<String, Object>> resultMapList =  dynamicTableMapper.select(taskDTO.getTable(),
                    taskDTO.getPrimary(),
                    String.valueOf(beginId));

            for (Map<String, Object> resultMap : resultMapList) {
                CdcDTO cdcDTO = CdcDTO.builder()
                        .type("INSERT")
                        .table(taskDTO.getTable())
                        .database("company")
                        .data(Collections.singletonList(resultMap))
                        .build();
                sendMsg(producer, taskDTO.getTopic(), taskDTO.getTable(), cdcDTO);
            }

            // 计算最大ID
            Long lastId = resultMapList.stream()
                    .map(map -> Long.parseLong(String.valueOf(map.get(taskDTO.getPrimary()))))
                    .max(Long::compareTo)
                    .orElse(null);
            log.info("最大的Id：{}-{}", taskDTO.getTable(), lastId);

            beginId = lastId;

        }

    }
}
